package com.jzl.canal_consumer.etl.load.mysql.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.jzl.canal_consumer.etl.load.mysql.DbDialect;
import com.jzl.canal_consumer.etl.load.mysql.SqlExecutor;
import com.jzl.canal_consumer.etl.load.mysql.SqlTemplate;
import com.jzl.canal_consumer.util.SqlUtils;
import com.jzl.util.LoggerUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.ddlutils.model.Column;
import org.apache.ddlutils.model.Table;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.StatementCreatorUtils;
import org.springframework.jdbc.support.lob.LobCreator;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author ZhenWuWang
 */
@Component
public class SqlExecutorImpl implements SqlExecutor
{
    private final DbDialect dbDialect;

    private final SqlTemplate sqlTemplate;

    @Autowired
    public SqlExecutorImpl(DbDialect dbDialect, SqlTemplate sqlTemplate)
    {
        this.dbDialect = dbDialect;
        this.sqlTemplate = sqlTemplate;
    }

    @Override
    public int[] batchExecute(CanalEntry.Header header, List<List<CanalEntry.Column>> columns)
    {
        if (columns == null || columns.isEmpty())
        {
            return new int[0];
        }
        long time = System.currentTimeMillis();
        String sql = getSql(header, columns.get(0));
        JdbcTemplate template = dbDialect.getJdbcTemplate();
        LobCreator lobCreator = dbDialect.getLobHandler().getLobCreator();
        int[] affects = template.batchUpdate(sql, new BatchPreparedStatementSetter()
        {
            @Override
            public void setValues(PreparedStatement preparedStatement, int i) throws SQLException
            {
                doPreparedStatement(preparedStatement, dbDialect, lobCreator, header, columns.get(i));
            }

            @Override
            public int getBatchSize()
            {
                return columns.size();
            }
        });
        int failureCount = 0;
        for (int affect : affects)
        {
            if (affect <= 0)
            {
                failureCount++;
            }
        }
        LoggerUtil.info(String.format("batchExecute sql:\n%s\n总记录数：%s,失败记录数：%s 耗时：%s",
                sql, columns.size(), failureCount, System.currentTimeMillis() - time));
        return affects;
    }

    @Override
    public int execute(CanalEntry.Header header, List<CanalEntry.Column> columns)
    {
        PlatformTransactionManager transactionManager = null;
        TransactionStatus status = null;
        try (LobCreator lobCreator = dbDialect.getLobHandler().getLobCreator())
        {
            // 定义事务属性
            DefaultTransactionDefinition transDef = new DefaultTransactionDefinition();
            // 设置传播行为属性
            transDef.setPropagationBehavior(DefaultTransactionDefinition.PROPAGATION_REQUIRED);
            // 获得事务状态
            transactionManager = dbDialect.getTransactionTemplate().getTransactionManager();
            if (transactionManager != null)
            {
                status = transactionManager.getTransaction(transDef);
                final String sql = getSql(header, columns);
                JdbcTemplate template = dbDialect.getJdbcTemplate();
                int affect = template.update(sql, ps -> doPreparedStatement(ps, dbDialect, lobCreator, header, columns));
                transactionManager.commit(status);
                return affect;
            }
        } catch (Exception e)
        {
            if (transactionManager != null && status != null)
            {
                transactionManager.rollback(status);
            }
            throw e;
        }
        return 0;
    }

    private String getSql(CanalEntry.Header header, List<CanalEntry.Column> columns)
    {
        List<String> pkNames = new ArrayList<>();
        List<String> colNames = new ArrayList<>();
        for (CanalEntry.Column column : columns)
        {

            if (column.getIsKey())
            {
                pkNames.add(column.getName());
            } else
            {
                if ("id".equals(column.getName()))
                {
                    pkNames.add(column.getName());
                } else
                {
                    colNames.add(column.getName());
                }
            }
        }
        String sql = "";
        CanalEntry.EventType eventType = header.getEventType();
        switch (eventType)
        {
            case INSERT:
                sql = sqlTemplate.getMergeInsertSql(header.getSchemaName(), header.getTableName(),
                        pkNames.toArray(new String[]{}), colNames.toArray(new String[]{}),
                        null, true);
                break;
            case UPDATE:
                sql = sqlTemplate.getUpdateSql(header.getSchemaName(), header.getTableName(),
                        pkNames.toArray(new String[]{}), colNames.toArray(new String[]{}));
                break;
            case DELETE:
                sql = sqlTemplate.getDeleteSql(header.getSchemaName(), header.getTableName(),
                        pkNames.toArray(new String[]{}));
            default:
                break;
        }
        return sql;
    }

    private void doPreparedStatement(PreparedStatement ps, DbDialect dbDialect, LobCreator lobCreator,
                                     CanalEntry.Header header, List<CanalEntry.Column> columns) throws SQLException
    {

        List<CanalEntry.Column> rebuildColumns = new ArrayList<>(columns.size());

        List<CanalEntry.Column> keyColumns = new ArrayList<>(columns.size());
        List<CanalEntry.Column> notKeyColumns = new ArrayList<>(columns.size());
        for (CanalEntry.Column column : columns)
        {
            if (column.getIsKey())
            {
                keyColumns.add(column);
            } else
            {
                notKeyColumns.add(column);
            }
        }
        CanalEntry.EventType eventType = header.getEventType();
        switch (eventType)
        {
            case INSERT:
            case UPDATE:
                // insert/update语句对应的字段数序都是将主键排在后面
                rebuildColumns.addAll(notKeyColumns);
                rebuildColumns.addAll(keyColumns);
                break;
            case DELETE:
                rebuildColumns.addAll(keyColumns);
            default:
                break;
        }

        // 获取一下当前字段名的数据是否必填
        Table table = dbDialect.findTable(header.getSchemaName(), header.getTableName());
        Map<String, Boolean> isRequiredMap = new HashMap();
        for (Column tableColumn : table.getColumns())
        {
            isRequiredMap.put(StringUtils.lowerCase(tableColumn.getName()), tableColumn.isRequired());
        }

        List<Object> values = new ArrayList<>(rebuildColumns.size());
        for (int i = 0; i < rebuildColumns.size(); i++)
        {
            int paramIndex = i + 1;
            CanalEntry.Column column = rebuildColumns.get(i);
            int sqlType = column.getSqlType();

            Boolean isRequired = isRequiredMap.get(StringUtils.lowerCase(column.getName()));
            if (isRequired == null)
            {
                // 清理一下目标库的表结构,二次检查一下
                table = dbDialect.findTable(header.getSchemaName(), header.getTableName());

                isRequiredMap = new HashMap<>();
                for (Column tableColumn : table.getColumns())
                {
                    isRequiredMap.put(StringUtils.lowerCase(tableColumn.getName()), tableColumn.isRequired());
                }

                isRequired = isRequiredMap.get(StringUtils.lowerCase(column.getName()));
                if (isRequired == null)
                {
                    throw new CanalClientException(String.format("column name %s is not found in Table[%s]",
                            column.getName(),
                            table.toString()));
                }
            }

            Object param;
            if (sqlType == Types.TIME || sqlType == Types.TIMESTAMP || sqlType == Types.DATE)
            {
                // 解决mysql的0000-00-00 00:00:00问题，直接依赖mysql
                // driver进行处理，如果转化为Timestamp会出错
                param = column.getValue();
                if (param != null && StringUtils.isEmpty(String.valueOf(param)))
                {
                    param = null;
                }
            } else
            {
                param = SqlUtils.stringToSqlValue(column.getValue(),
                        sqlType,
                        isRequired,
                        column.getIsNull());
            }

            try
            {
                switch (sqlType)
                {
                    case Types.CLOB:
                        lobCreator.setClobAsString(ps, paramIndex, (String) param);
                        break;
                    case Types.BLOB:
                        lobCreator.setBlobAsBytes(ps, paramIndex, (byte[]) param);
                        break;
                    case Types.TIME:
                    case Types.TIMESTAMP:
                    case Types.DATE:
                        ps.setObject(paramIndex, param);
                        break;
                    case Types.BIT:
                        StatementCreatorUtils.setParameterValue(ps, paramIndex, Types.DECIMAL, null, param);
                        break;
                    default:
                        StatementCreatorUtils.setParameterValue(ps, paramIndex, sqlType, null, param);
                        break;
                }
                values.add(param);
            } catch (SQLException ex)
            {
                LoggerUtil.error("## SetParam error , [sqltype={}, value={}]", sqlType, param);
                throw ex;
            }
        }
        LoggerUtil.debug("## sql values: {}", JSON.toJSONString(values));
    }
}
